#ifndef __M_CHANNEL_H__
#define __M_CHANNEL_H__
#include "muduo/net/TcpConnection.h"
#include "muduo/proto/codec.h"
#include "muduo/proto/dispatcher.h"
#include "../mqcommon/logger.hpp"
#include "../mqcommon/helper.hpp"
#include "../mqcommon/mq_msg.pb.h"
#include "../mqcommon/mq_proto.pb.h"
#include "../mqcommon/threadpool.hpp"
#include "consumer.hpp"
#include <iostream>
#include <mutex>
#include <condition_variable>
#include <unordered_map>

using ProtobufCodecPtr = std::shared_ptr<ProtobufCodec>;
using basicConsumeResponsePtr = std::shared_ptr<mymq::basicConsumeResponse>;
using basicCommonResponsePtr = std::shared_ptr<mymq::basicCommonResponse>;

class Channel{
public:
    using ptr = std::shared_ptr<Channel>;

    Channel(const muduo::net::TcpConnectionPtr &conn, const ProtobufCodecPtr &codec):
        _cid(UUIDHelper::uuid()), _conn(conn), _codec(codec){}
    ~Channel(){
        basicCancel();//取消订阅
    }

    std::string cid(){
        return _cid;
    }

    bool openChannel(){
        std::string rid = UUIDHelper::uuid();
        mymq::openChannelRequest req;
        req.set_rid(rid);
        req.set_cid(_cid);
        _codec->send(_conn, req);
        basicCommonResponsePtr resp = waitResponse(rid);
        return resp->ok();
    }

    void closeChannel(){
        std::string rid = UUIDHelper::uuid();
        mymq::closeChannelRequest req;
        req.set_rid(rid);
        req.set_cid(_cid);
        _codec->send(_conn, req);
        waitResponse(rid);
        return ;
    }

    bool declareExchange(const std::string &ename, mymq::ExchangeType etype, bool edurable,
                        bool eauto_delete, google::protobuf::Map<std::string, std::string> &eargs){
        //构造一个声明交换机的请求对象
        std::string rid = UUIDHelper::uuid();
        mymq::declareExchangeRequest req;
        req.set_rid(rid);//请求id
        req.set_cid(_cid);//信道id
        req.set_exchange_name(ename);
        req.set_exchange_type(etype);
        req.set_durable(edurable);
        req.set_auto_delete(eauto_delete);
        req.mutable_args()->swap(eargs);
        //向服务器发送请求
        _codec->send(_conn, req);
        //等待服务器响应
        basicCommonResponsePtr resp = waitResponse(rid);
        //返回
        return resp->ok();
    }

    void deleteExchange(const std::string &ename){
        //操作都雷同
        std::string rid = UUIDHelper::uuid();
        mymq::deleteExchangeRequest req;
        req.set_rid(rid);
        req.set_cid(_cid);
        req.set_exchange_name(ename);
        _codec->send(_conn, req);
        waitResponse(rid);
        return;
    }

    bool declareQueue(const std::string &qname, bool qdurable, bool qexclusive,
                    bool qauto_delete, google::protobuf::Map<std::string, std::string> &qargs){
        std::string rid = UUIDHelper::uuid();
        mymq::declareQueueRequest req;
        req.set_rid(rid);
        req.set_cid(_cid);
        req.set_queue_name(qname);
        req.set_durable(qdurable);
        req.set_auto_delete(qauto_delete);
        req.set_exclusive(qexclusive);
        req.mutable_args()->swap(qargs);
        _codec->send(_conn, req);
        basicCommonResponsePtr resp = waitResponse(rid);
        return resp->ok();
    }

    void deleteQueue(const std::string &qname){
        std::string rid = UUIDHelper::uuid();
        mymq::deleteQueueRequest req;
        req.set_rid(rid);
        req.set_cid(_cid);
        req.set_queue_name(qname);
        _codec->send(_conn, req);
        waitResponse(rid);
        return;
    }

    bool queueBind(const std::string &ename, const std::string &qname, const std::string &key){
        std::string rid = UUIDHelper::uuid();
        mymq::queueBindRequest req;
        req.set_rid(rid);
        req.set_cid(_cid);
        req.set_exchange_name(ename);
        req.set_queue_name(qname);
        req.set_binding_key(key);
        _codec->send(_conn, req);
        basicCommonResponsePtr resp = waitResponse(rid);
        return resp->ok();
    }
    
    void queueUnBind(const std::string &ename, const std::string &qname){
        std::string rid = UUIDHelper::uuid();
        mymq::queueUnBindRequest req;
        req.set_rid(rid);
        req.set_cid(_cid);
        req.set_exchange_name(ename);
        req.set_queue_name(qname);
        _codec->send(_conn, req);
        waitResponse(rid);
        return;
    }

    void basicPublish(const std::string &ename, const mymq::BasicProperties *bp, const std::string &body){
        std::string rid = UUIDHelper::uuid();
        mymq::basicPublishRequest req;
        req.set_rid(rid);
        req.set_cid(_cid);
        req.set_exchange_name(ename);
        req.set_body(body);
        if (bp != nullptr){
            req.mutable_properties()->set_id(bp->id());
            req.mutable_properties()->set_delivery_mode(bp->delivery_mode());
            req.mutable_properties()->set_routing_key(bp->routing_key());
        }
        _codec->send(_conn, req);
        waitResponse(rid);
        return ;
    }
    
    void basicAck(const std::string &msgid){
        if (_consumer.get() == nullptr){
            DLOG("消息确认时，找不到消费者信息！");
            return ;
        }
        std::string rid = UUIDHelper::uuid();
        mymq::basicAckRequest req;
        req.set_rid(rid);
        req.set_cid(_cid);
        req.set_queue_name(_consumer->qname);
        req.set_message_id(msgid);
        _codec->send(_conn, req);
        waitResponse(rid);
        return ;
    }

    bool basicConsume(const std::string &consumer_tag, const std::string &queue_name,
                    bool auto_ack, const ConsumerCallback &cb){
        if(_consumer.get() != nullptr){
            DLOG("当前信道已订阅其他队列消息！");
            return false;
        }
        std::string rid = UUIDHelper::uuid();
        mymq::basicConsumeRequest req;
        req.set_rid(rid);
        req.set_cid(_cid);
        req.set_consumer_tag(consumer_tag);
        req.set_queue_name(queue_name);
        req.set_auto_ack(auto_ack);
        _codec->send(_conn, req);

        basicCommonResponsePtr resp = waitResponse(rid);
        if (resp->ok() == false){
            DLOG("添加订阅失败！");
            return false;
        }
        _consumer = std::make_shared<Consumer>(consumer_tag, queue_name, auto_ack, cb);
        return true;
    }

    void basicCancel(){
        if(_consumer.get() == nullptr){
            return ;
        }
        std::string rid = UUIDHelper::uuid();
        mymq::basicCancelRequest req;
        req.set_rid(rid);
        req.set_cid(_cid);
        req.set_queue_name(_consumer->qname);
        req.set_consumer_tag(_consumer->tag);
        _codec->send(_conn, req);
        waitResponse(rid);
        _consumer.reset();
        return ;
    }

public:
    //连接收到基础相应后，向hash_map中添加响应
    void pushBasicResponse(const basicCommonResponsePtr& resp){
        std::unique_lock<std::mutex> lock(_mutex);
        _basic_resp.insert(std::make_pair(resp->rid(), resp));
        _cv.notify_all();//唤醒所有阻塞，其实就是waitResponse里的阻塞
    }
    //连接收到消息推送后，需要通过信道找到对应的消费者对象，通过回调函数进行消息处理
    void consume(const basicConsumeResponsePtr &resp){
        if (_consumer.get() == nullptr){
            DLOG("消息处理时，未找到订阅者信息!");
            return ;
        }
        if (_consumer->tag != resp->consumer_tag()){
            DLOG("收到的推送消息中的消费者标识，与当前信道消费者标识不匹配！");
            return ;
        }
        _consumer->callback(resp->consumer_tag(), resp->mutable_properties(), resp->body());
    }

private:
    basicCommonResponsePtr waitResponse(const std::string &rid){
        std::unique_lock<std::mutex> lock(_mutex);
        _cv.wait(lock, [&rid, this](){
            return _basic_resp.find(rid) != _basic_resp.end();
        });
        basicCommonResponsePtr basic_resp = _basic_resp[rid];
        _basic_resp.erase(rid);
        return basic_resp;
    }
private:
    std::string _cid;
    muduo::net::TcpConnectionPtr _conn;
    ProtobufCodecPtr _codec;
    Consumer::ptr _consumer;
    std::mutex _mutex;
    std::condition_variable _cv;
    std::unordered_map<std::string, basicCommonResponsePtr> _basic_resp;
};

//信道管理-增删查
class ChannelManager{
public:
    using ptr = std::shared_ptr<ChannelManager>;
    ChannelManager(){}

    Channel::ptr create(const muduo::net::TcpConnectionPtr &conn, const ProtobufCodecPtr &codec){
        std::unique_lock<std::mutex> lock(_mutex);
        auto channel = std::make_shared<Channel>(conn, codec);
        _channels.insert(std::make_pair(channel->cid(), channel));
        return channel;
    }

    void remove(const std::string &cid){
        std::unique_lock<std::mutex> lock(_mutex);
        _channels.erase(cid);
    }

    Channel::ptr get(const std::string &cid){
        std::unique_lock<std::mutex> lock(_mutex);
        auto it = _channels.find(cid);
        if (it == _channels.end()){
            return Channel::ptr();
        }
        return it->second;
    }
private:
    std::mutex _mutex;
    std::unordered_map<std::string, Channel::ptr> _channels;
};
#endif